--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit c7fe6df4d772b84e4231daf036ca127ed1392632
Parents : 6508edf
Author : Ivan <ivan@quad4.io>
Signature : Signature validation error
Date : 2026-04-22T20:09:47-05:00
feat(lxmf): add LXMF sieve filter implementation for message and peer matching
Changes
Diff
diff --git a/meshchatx/src/backend/lxmf_sieve.py b/meshchatx/src/backend/lxmf_sieve.py
new file mode 100644
index 00000000..20fc8229
--- /dev/null
+++ b/meshchatx/src/backend/lxmf_sieve.py
@@ -0,0 +1,215 @@
+# SPDX-License-Identifier: 0BSD
+
+"""LXMF sieve filters: match peer/message text (substring or regex) and apply actions."""
+
+from __future__ import annotations
+
+import json
+import re
+import uuid
+from typing import Any
+
+MAX_RULES = 64
+MAX_TERMS_PER_RULE = 32
+MAX_TERM_LEN = 512
+
+SIEVE_SCOPES = frozenset({"everyone", "contacts", "non_contacts"})
+SIEVE_ACTIONS = frozenset({"hide", "ignore", "folder", "banish"})
+MATCH_MODES = frozenset({"substring", "regex"})
+
+_REGEX_FLAGS = re.IGNORECASE | re.DOTALL
+
+
+def _new_rule_id() -> str:
+ return uuid.uuid4().hex[:16]
+
+
+def _validate_regex_terms(terms: list[str]) -> list[str] | None:
+ out: list[str] = []
+ for raw in terms:
+ p = str(raw)[:MAX_TERM_LEN]
+ if not p.strip():
+ continue
+ try:
+ re.compile(p, _REGEX_FLAGS)
+ except re.error:
+ return None
+ out.append(p)
+ return out if out else None
+
+
+def _normalize_action(raw: str | None) -> str | None:
+ if raw == "block":
+ return "hide"
+ if raw in SIEVE_ACTIONS:
+ return raw
+ return None
+
+
+def normalize_lxmf_sieve_filters(filters: list) -> list[dict[str, Any]]:
+ """Validate and normalize a list of sieve rule dicts from user input."""
+ out: list[dict[str, Any]] = []
+ if not isinstance(filters, list):
+ return out
+ for item in filters[:MAX_RULES]:
+ if not isinstance(item, dict):
+ continue
+ action = _normalize_action(item.get("action"))
+ if action is None:
+ continue
+
+ match_mode = item.get("match_mode") or "substring"
+ if match_mode not in MATCH_MODES:
+ match_mode = "substring"
+
+ match_peer_fields = item.get("match_peer_fields")
+ if match_peer_fields is None:
+ match_peer_fields = True
+ else:
+ match_peer_fields = bool(match_peer_fields)
+ match_message = bool(item.get("match_message", False))
+ if not match_peer_fields and not match_message:
+ continue
+
+ terms_in = item.get("terms")
+ if isinstance(terms_in, str):
+ terms = [t.strip() for t in re.split(r"[\n,]+", terms_in) if t.strip()]
+ elif isinstance(terms_in, list):
+ terms = [str(t).strip() for t in terms_in if str(t).strip()]
+ else:
+ terms = []
+ terms = [t[:MAX_TERM_LEN] for t in terms[:MAX_TERMS_PER_RULE]]
+
+ if match_mode == "regex":
+ validated = _validate_regex_terms(terms)
+ if validated is None:
+ continue
+ terms = validated
+ else:
+ if not terms:
+ continue
+
+ folder_id: int | None = None
+ if action == "folder":
+ try:
+ folder_id = int(item.get("folder_id"))
+ except (TypeError, ValueError):
+ continue
+
+ rid = str(item.get("id") or "").strip() or _new_rule_id()
+ scope = item.get("scope") or item.get("applies_to")
+ if scope not in SIEVE_SCOPES:
+ scope = "everyone"
+ out.append(
+ {
+ "id": rid,
+ "enabled": bool(item.get("enabled", True)),
+ "scope": scope,
+ "match_peer_fields": match_peer_fields,
+ "match_message": match_message,
+ "match_mode": match_mode,
+ "terms": terms,
+ "action": action,
+ "folder_id": folder_id,
+ },
+ )
+ return out
+
+
+def parse_lxmf_sieve_filters_json(raw: str | None) -> list[dict[str, Any]]:
+ if not raw or not str(raw).strip():
+ return []
+ try:
+ data = json.loads(raw)
+ except (json.JSONDecodeError, TypeError):
+ return []
+ if not isinstance(data, list):
+ return []
+ return normalize_lxmf_sieve_filters(data)
+
+
+def _rule_scope_matches(rule: dict[str, Any], is_contact: bool) -> bool:
+ scope = rule.get("scope") or "everyone"
+ if scope not in SIEVE_SCOPES:
+ scope = "everyone"
+ if scope == "everyone":
+ return True
+ if scope == "contacts":
+ return is_contact
+ if scope == "non_contacts":
+ return not is_contact
+ return True
+
+
+def _any_term_matches_substring(terms: list[str], haystack_lower: str) -> bool:
+ hs = haystack_lower or ""
+ for term in terms:
+ t = str(term).lower()
+ if t and t in hs:
+ return True
+ return False
+
+
+def _any_term_matches_regex(terms: list[str], haystack_raw: str) -> bool:
+ text = haystack_raw or ""
+ for pattern in terms:
+ try:
+ if re.search(pattern, text, _REGEX_FLAGS):
+ return True
+ except re.error:
+ continue
+ return False
+
+
+def _terms_match_rule_on_text(
+ rule: dict[str, Any],
+ text_lower: str,
+ text_raw: str,
+) -> bool:
+ terms = rule.get("terms") or []
+ mode = rule.get("match_mode") or "substring"
+ if mode == "regex":
+ return _any_term_matches_regex(terms, text_raw)
+ return _any_term_matches_substring(terms, text_lower)
+
+
+def first_matching_lxmf_sieve_rule(
+ rules: list[dict[str, Any]],
+ peer_haystack: str | None,
+ *,
+ is_contact: bool = False,
+ message_haystack: str | None = None,
+) -> dict[str, Any] | None:
+ """Return the first enabled rule whose scope and match flags are satisfied."""
+ peer_raw = peer_haystack or ""
+ peer_lower = peer_raw.lower()
+ msg_raw = message_haystack
+ msg_lower = (message_haystack or "").lower()
+
+ for rule in rules:
+ if not rule.get("enabled", True):
+ continue
+ if not _rule_scope_matches(rule, is_contact):
+ continue
+
+ peer_fields = rule.get("match_peer_fields", True)
+ match_msg = rule.get("match_message", False)
+
+ peer_ok = True
+ if peer_fields:
+ peer_ok = _terms_match_rule_on_text(rule, peer_lower, peer_raw)
+
+ msg_ok = True
+ if match_msg:
+ if msg_raw is None:
+ continue
+ msg_ok = _terms_match_rule_on_text(rule, msg_lower, msg_raw)
+
+ if peer_ok and msg_ok:
+ return {
+ "action": rule["action"],
+ "folder_id": rule.get("folder_id"),
+ "rule_id": rule.get("id"),
+ "scope": rule.get("scope") or "everyone",
+ }
+ return None
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────